package com.gitee.huanminabc.jcommon.container;

import java.util.AbstractList;
import java.util.ArrayList;
import java.util.List;
import java.util.RandomAccess;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Consumer;

/**
 * List分片处理,每次处理size个元素,支持每个分片异步处理
 * 如果使用多线程模式,那么不要在事务中使用,因为每个分片是一个新的线程,会导致事务失效
 * @author huanmin
 * @date 2021/1/6
 */
public class ListSlice {

    /**
     * 将一个列表分割成多个子列表
     * @param list 要分割的列表
     * @param size 每个子列表的大小
     * @return 分割后的子列表
     */
    public static <T> List<List<T>> slice(List<T> list, int size) {
        return (list instanceof RandomAccess)
                ? new RandomAccessPartition<T>(list, size)
                : new Partition<T>(list, size);
    }


    //将list分片,每片size个元素,然后对每片元素执行consumer
    public static <T> List<Consumer<List<T>>> slice(List<T> list,int size, Consumer<List<T>> consumer) {
        List<Consumer<List<T>>> consumers = new ArrayList<>();
        List<List<T>> partition = slice(list, size);
        if (partition.isEmpty()) {
            return consumers;
        }
        for (List<T> ts : partition) {
            consumers.add(t -> consumer.accept(ts));
        }
        return consumers;
    }

    //异步执行
    public static <T>  void sliceAsyncRun(List<T> list,int size, Consumer<List<T>> consumer) {
        //获取cpu核心数
        int processors = Runtime.getRuntime().availableProcessors()*2;
        sliceAsyncRun(list, size, consumer, processors);
    }
    //异步执行,定义流的线程数

    /**
     * 异步执行
     * @param list 数据
     * @param size 分片大小
     * @param consumer 消费者 对每个分片的数据进行处理
     * @param threadNum 线程数也就是同时并行处理的分片数,  如果是io密集型任务,线程数可以设置大一点,只要cpu和内存能承受
     * @param <T>
     */
    public static <T>  void sliceAsyncRun(List<T> list,int size, Consumer<List<T>> consumer,int threadNum) {
        List<Consumer<List<T>>> slice = slice(list, size, consumer);
        //获取cpu核心数
        int processors = Runtime.getRuntime().availableProcessors()*2;
        ForkJoinPool customThreadPool = new ForkJoinPool(Math.max(processors, threadNum));
        customThreadPool.submit(() -> slice.parallelStream().forEach(t -> new Thread(() -> t.accept(list)).start()));
    }


    //同步
    public  static <T>  void sliceSyncRun(List<T> list,int size, Consumer<List<T>> consumer){
        List<Consumer<List<T>>> slice = slice(list, size, consumer);
        slice.forEach(t -> t.accept(list));
    }


    private static class Partition<T> extends AbstractList<List<T>> {
        final List<T> list;
        final int size;

        Partition(List<T> list, int size) {
            this.list = list;
            this.size = size;
        }

        @Override
        public List<T> get(int index) {
            int start = index * size;
            int end = Math.min(start + size, list.size());
            return list.subList(start, end);
        }

        @Override
        public int size() {
            return (int) Math.ceil((double) list.size() / size);
        }

        @Override
        public boolean isEmpty() {
            return list.isEmpty();
        }
    }
    private static class RandomAccessPartition<T> extends Partition<T> implements RandomAccess {
        RandomAccessPartition(List<T> list, int size) {
            super(list, size);
        }
    }

}
